package com.frank.demo.zookeeper.curator.zk;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.commons.lang3.StringUtils;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.NodeCache;
import org.apache.curator.framework.recipes.cache.NodeCacheListener;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;

/**
 * 分布式锁
 * 主要负责选主，分配任务/节点
 * Created by zhy on 2018/5/2.
 */
public class LockHandler extends AbstractZkHandler {
    private static final Logger log = LoggerFactory.getLogger(LockHandler.class);
    private int min_node = 2;
    private Boolean MASTER = false;
    private Boolean RUNNING = false;
    private LockStart lockStart;

    public LockHandler(String path, String zkAddress, int min_node, LockStart lockStart) {
        this.path = path;
        this.zkAddress = zkAddress;
        this.min_node = min_node;
        this.lockStart = lockStart;
    }

    @Override
    public void routeChange(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
        nodeChange(client, event);
    }


    @Override
    public void routeDataUpdate(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
        nodeChange(client, event);
    }


    @Override
    public NodeCacheListener getNodeChange(final NodeCache nodeCache) {
        NodeCacheListener nodeCacheListener = new NodeCacheListener() {
            @Override
            public void nodeChanged() throws Exception {
                String str = new String(nodeCache.getCurrentData().getData(), "UTF-8");
                JSONObject data = JSONObject.parseObject(str);
                if (data.getIntValue("STATUS") == 1 && data.getString("RANGES") != null) {
                    System.out.println("SLEF 数据变更,启动");
                    RUNNING = true;
                    lockStart.onStart(data);
                }
            }
        };

        return nodeCacheListener;
    }

    @Override
    public void shutDown() throws Exception {
        lockStart.onDestory();
    }

    /**
     * master节点工作,分配数据
     *
     * @param children
     * @param client
     * @throws Exception
     */
    private void masterJob(List<String> children, CuratorFramework client) throws Exception {
        List<String> allRangeList = getAllQueueRanges();
        List<String> rangeList = allocationBatchNode(allRangeList, min_node);

        //遍历去除已经存在的节点
        for (String child : children) {
            // 修改为String便于查看
            String data = ZkStarter.getData(client, path + "/" + child);
            if (data != null) {
                JSONObject mdata = JSON.parseObject(data);
                String groupRange = (String) mdata.get("RANGES");
                if (groupRange != null) {
                    rangeList.remove(groupRange);
                }
            }
        }

        //遍历设置剩余节点
        for (String child : children) {
            String data = ZkStarter.getData(client, path + "/" + child);
            if (data != null) {
                JSONObject mdata = JSON.parseObject(data);

                String groupRange = (String) mdata.get("RANGES");
                if (StringUtils.isBlank(groupRange)) {
                    groupRange = rangeList.get(0);
                    mdata.put("RANGES", groupRange);
                    ZkStarter.setData(client, path + "/" + child, mdata);
                    rangeList.remove(groupRange);
                }

                if (rangeList.isEmpty()) {
                    break;
                }
            }
        }
    }

    private void nodeChange(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
        List<String> children = client.getChildren().forPath(path);
        Collections.sort(children);
        int index = children.indexOf(self);

        checkNode(client, index);

        //leader 节点
        if (MASTER) {
            masterJob(children, client);
        } else if (index == -1) {
            shutDown();
        }
    }

    private void checkNode(CuratorFramework client, int index) throws Exception {
        try {
            String data = ZkStarter.getData(client, selfPath);
            JSONObject noda_data = JSONObject.parseObject(data);
            if (noda_data == null || noda_data.getInteger("STATUS") == null) {
                shutDown();
            }
            if (index == 0 && !MASTER) {
                noda_data.put("MASTER", true);
                MASTER = true;
                ZkStarter.setData(client, selfPath, noda_data);
            }
        } catch (Exception e) {
            log.error("获取zk node数据失败");
            shutDown();
        }
    }

    /**
     * 分配批处理节点
     *
     * @param allRangeList
     * @param nodes
     * @return
     */
    private List<String> allocationBatchNode(List<String> allRangeList, int nodes) {
        List<String> rev = new ArrayList();
        List<String> pdata = new ArrayList();
        List<String> gdata = new ArrayList();
        for (String queue : allRangeList) {
            if (StringUtils.startsWith(queue, "GROUP"))
                gdata.add(queue);
            else
                pdata.add(queue);
        }

        if (nodes <= 0 || allRangeList.size() == 0) return rev;

        List<String> grev = new ArrayList();
        if (gdata.size() <= nodes) {
            grev.addAll(gdata);
        } else {
            int floor = gdata.size() / nodes;
            int mod = gdata.size() - nodes * floor; //data.size()%nodes;

            int modIndex = 0, index = 0;
            for (int i = 0; i < nodes; i++) {
                StringBuilder sb = new StringBuilder();
                int max = floor;
                if (modIndex < mod) {
                    max += 1;
                }
                for (int j = 0; j < max; j++) {
                    if (sb.length() > 1) sb.append(",");
                    sb.append(gdata.get(index++));
                }
                grev.add(sb.toString());
                modIndex++;
            }
        }

        List<String> prev = new ArrayList();
        if (pdata.size() <= nodes) {
            prev.addAll(pdata);
        } else {
            int floor = pdata.size() / nodes;
            int mod = pdata.size() - nodes * floor; //data.size()%nodes;

            int modIndex = 0;
            int index = 0;
            for (int i = 0; i < nodes; i++) {
                StringBuilder sb = new StringBuilder();
                int max = floor;
                if (modIndex < mod) {
                    max += 1;
                }
                for (int j = 0; j < max; j++) {
                    if (sb.length() > 0) sb.append(",");
                    sb.append(pdata.get(index++));
                }
                prev.add(sb.toString());
                modIndex++;
            }
        }

        if (prev.size() + grev.size() <= nodes) {
            rev.addAll(grev);
            rev.addAll(prev);
        } else {
            int pStart = nodes - prev.size();
            for (int i = 0; i < nodes; i++) {
                String range = null;
                if (grev.size() > i)
                    range = grev.get(i);
                if (i - pStart >= 0) {
                    if (range != null) {
                        range += ",";
                        range += prev.get(i - pStart);
                    } else {
                        range = prev.get(i - pStart);
                    }
                }
                if (range != null)
                    rev.add(range);
            }
        }
        return rev;
    }

    private List<String> getAllQueueRanges() {
        String _strQueueRanges = "64,GROUP00";
        String[] q = StringUtils.split(_strQueueRanges, ",");
        Integer pQueueNum = Integer.valueOf(q[0]);//个人帐户队列数
        List<String> rangeList = new ArrayList();
        for (int i = 0; i < pQueueNum; i++) {
            rangeList.add(String.valueOf(i));
        }

        for (int i = 1; i < q.length; i++) {
            rangeList.add(q[i]);
        }
        return rangeList;
    }

    public interface LockStart {
        void onStart(JSONObject data);

        void onDestory() throws Exception;
    }

}